package com.uptown.matrix.stream.sink;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import com.alibaba.fastjson.JSON;
import com.uptown.matrix.stream.builder.RedisTaskBuilder;
import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisFuture;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;


/**
 * sink消费规则
 */
@Slf4j
public class MatrixStreamSink implements SinkFunction<Map> {

    RedisClient redisClient;

    public MatrixStreamSink(RedisClient redisClient) {
        this.redisClient = redisClient;
    }

    @Override
    public void invoke(Map taskInfo, Context context) throws Exception {
        System.out.println();
        RedisTaskBuilder.pipeline(this.redisClient, redisAsyncCommands -> {

            List<RedisFuture<?>> redisFutures = new ArrayList<>();
//            redisFutures.add(redisAsyncCommands.lpush(id.getBytes(), JSON.toJSONString(simpleAnchorInfo).getBytes()));
            return redisFutures;
        });
    }


}
